import abc
import multiprocessing
import queue
import threading
import time
from typing import Any

import grpc

from ys_python.common.base.logger import get_logger_with_cfg, LoggerCfg
from ys_python.proto import MQ_pb2_grpc, MQ_pb2


class IYSQueue(abc.ABC):
    """
    自定义队列虚类
    """

    def __init__(self, id_: int):
        self.id = id_

    @property
    def id(self) -> int:
        """
         唯一标识
        """
        return self._id

    @id.setter
    def id(self, value):
        self._id = value

    @abc.abstractmethod
    def get(self) -> Any:
        """
        阻塞读取
        :return: 返回读取到的数据
        """
        pass

    @abc.abstractmethod
    def try_get(self) -> Any:
        """
        非阻塞读取
        :return: 返回 None 为没有读取到数据，否则返回 数据
        """
        pass

    @abc.abstractmethod
    def put(self, msg):
        """
        阻塞读取
        :return: 返回读取到的数据
        """
        pass

    @abc.abstractmethod
    def try_put(self, msg) -> bool:
        """
        非阻塞读取
        :return: 返回 None 为没有读取到数据，否则返回 数据
        """
        pass

    def __cmp__(self, other) -> int:
        if self.id == other.id:
            return 0
        elif self.id > other.id:
            return 1
        else:
            return -1

    def __del__(self):
        pass


class SimpleYSQueue(IYSQueue):
    def put(self, msg):
        self.__queue.put(msg)

    def try_put(self, msg) -> bool:
        try:
            self.__queue.put_nowait(msg)
            return True
        except queue.Full as e:
            return False

    def __init__(self, id_: int, size=10):
        super().__init__(id_)
        self.__queue = multiprocessing.Queue(size)

    def get(self) -> Any:
        return self.__queue.get()

    def try_get(self) -> Any:
        try:
            return self.__queue.get_nowait()
        except queue.Empty as e:
            return None

    def __del__(self):
        while not self.__queue.empty():
            self.__queue.get()


class GRPCYSQueue(IYSQueue):
    """
    基于 GRPC 的 队列实现
    """

    __stream: Any

    def __get_data(self):
        while True:
            try:
                __stream = self.__sub.Sub(MQ_pb2.SubRequest(topics=self.__topics))
                for res in __stream:
                    try:
                        self.__queue.put_nowait(res)
                        self.__logger.debug(
                            "({}):接收消息 \n topic:{} \n message:{}".format(self.id, res.topic,
                                                                             str(res.message, "utf-8")))
                    except queue.Full:
                        self.__logger.warning("({}):订阅队列已满数据丢失!".format(self.id))

                self.__logger.info("订阅正常取消！")
                self.__queue.close()
                return
            except grpc.RpcError as e:
                if self.__is_del:
                    return
                self.id = -1
                self.__logger.warning(f"({self.id})无法连接, 2 秒后重试\n ")
                self.__logger.debug(f"故障类型：{e}")
                time.sleep(2)

    def __init__(self, id_: int, sub: MQ_pb2_grpc.MQStub, topics: str, logger_cfg: LoggerCfg = LoggerCfg(name="app"),
                 size=10):
        """
        :param topics: 订阅的话题 逗号分割
        """

        super().__init__(id_)
        self.__logger = get_logger_with_cfg(logger_cfg, "queue")
        self.__sub = sub
        self.__topics = topics
        self.__queue = multiprocessing.Queue(size)
        self.__is_del = False

        self.__logger.info("""
        ---队列初始化---
        订阅话题: {}
        大小: {}
        """.format(topics, size))

        # 启动接收线程获取数据
        self.__thread = threading.Thread(target=self.__get_data)
        self.__thread.start()

    def put(self, msg):
        pass

    def try_put(self, msg) -> bool:
        pass

    def get(self) -> Any:
        return self.__queue.get()

    def try_get(self) -> Any:
        try:
            return self.__queue.get_nowait()
        except queue.Empty as e:
            return None

    def __del__(self):
        self.__logger.debug(f"销毁队列{self.id}")
        try:
            self.__is_del = True
            self.__thread.join()
            while not self.__queue.empty():
                self.__queue.get()
            self.__queue.close()
        except Exception as e:
            return
